package main

import (
	"fmt"
	"github.com/sirupsen/logrus"
	"gopkg.in/ini.v1"
	"logagent/common"
	"logagent/etcd"
	"logagent/kafka"
	"logagent/tailfile"
)

//整个Logagent的配置结构体
type Config struct {
	KafkaConfig   `ini:"kafka"`
	CollectConfig `ini:"collect"`
	EtcdConfig    `ini:"etcd"`
}

type KafkaConfig struct {
	Address  string `ini:"address"`
	ChanSize int64  `ini:"chan_size"`
}

type CollectConfig struct {
	LogfilePath string `ini:"logfile_path"`
}

type EtcdConfig struct {
	Address    string `ini:"address"`
	CollectKey string `ini:"collect_key"`
}

//真正的业务逻辑
func run() {
	select {}
}
func main() {
	//-1.获取本机IP,为后续去etcd取配置文件打下基础
	ip, err := common.GetOutboundIP()
	if err != nil {
		logrus.Errorf("get ip failed,err:%v", err)
	}
	//0.第一种方式:读配置文件
	//cfg, err := ini.Load("./conf/config.ini")

	//0.第二种方式:读配置文件
	var configObj = new(Config)
	err = ini.MapTo(configObj, "./conf/config.ini")
	//0.1
	if err != nil {
		logrus.Error("load config failed,err:", err)
		return
	}
	fmt.Printf("%#v \n", configObj)
	//1.初始化连接kafka（做好准备工作）
	err = kafka.Init([]string{configObj.KafkaConfig.Address}, configObj.ChanSize)
	fmt.Printf("kafka：%v %d \n", configObj.KafkaConfig.Address, configObj.KafkaConfig.ChanSize)
	if err != nil {
		logrus.Error("init kafka failed,err:", err)
		return
	}
	logrus.Error("init kafka success!!!")

	//初始化etcd连接
	err = etcd.Init([]string{configObj.EtcdConfig.Address})
	if err != nil {
		logrus.Error("init etcd failed,err:", err)
		return
	}
	//从etcd拉取要收集日志的配置项
	collectKey := fmt.Sprintf(configObj.CollectKey, ip)
	fmt.Println("collectKey:", collectKey)
	allConf, err := etcd.GetConf(collectKey)
	if err != nil {
		logrus.Error("get conf from etcd failed,err:", err)
		return
	}
	fmt.Printf("allConf: %#v\n", allConf)
	//派一个小弟去监控etcd中configObj.CollectKey对应值的变化
	go etcd.WatchConf(collectKey)
	//2.根据配置中的日志路径使用tail去收集日志,初始化tail
	err = tailfile.Init(allConf) //把从etcd中获取的配置项传到Init中
	//fmt.Printf("%#v \n", configObj.LogfilePath)
	if err != nil {
		logrus.Error("init tailfile failed,err:", err)
		return
	}
	logrus.Error("init tailfile success!!!")
	//3.把日志通过sarama发往kafka
	//3.1从TailObj读取数据 TailObj->log->Client->kafka
	run()
}
